package com.atguigu.day04;

import com.atguigu.bean.OrderEvent;
import com.atguigu.bean.TxEvent;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.streaming.api.datastream.ConnectedStreams;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.co.KeyedCoProcessFunction;
import org.apache.flink.util.Collector;

import java.util.HashMap;

public class Flink05_Project_Order {
    public static void main(String[] args) throws Exception {
        //1.获取流的执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        env.setParallelism(1);

        //2.分别从csv文件中获取订单表的数据，和交易表的数据
        SingleOutputStreamOperator<OrderEvent> orderEventStream = env.readTextFile("input/OrderLog.csv")
                .map(new MapFunction<String, OrderEvent>() {
                    @Override
                    public OrderEvent map(String value) throws Exception {
                        String[] split = value.split(",");
                        return new OrderEvent(Long.parseLong(split[0]), split[1], split[2], Long.parseLong(split[3]));
                    }
                });

        SingleOutputStreamOperator<TxEvent> txEventStream = env.readTextFile("input/ReceiptLog.csv")
                .map(new MapFunction<String, TxEvent>() {
                    @Override
                    public TxEvent map(String value) throws Exception {
                        String[] split = value.split(",");
                        return new TxEvent(split[0], split[1], Long.parseLong(split[2]));
                    }
                });

        //3.使用Connect将两条流的数据连接起来成为一条流
        ConnectedStreams<OrderEvent, TxEvent> connect = orderEventStream.connect(txEventStream);

        //4.将相同交易码的数据聚合到一块
        ConnectedStreams<OrderEvent, TxEvent> keyBy = connect.keyBy("txId", "txId");

        //5.实时对账
        keyBy.process(new KeyedCoProcessFunction<String, OrderEvent, TxEvent, String>() {
            //创建缓存订单表数据的Map集合
            HashMap<String, OrderEvent> orderMap = new HashMap<>();

            //创建缓存交易表数据的Map集合
            HashMap<String, TxEvent> txMap = new HashMap<>();

            @Override
            public void processElement1(OrderEvent value, Context ctx, Collector<String> out) throws Exception {
                //1.判断对方缓存中是否有这条数据(orderEvent)能关联上的数据(TxEvent)
                if (txMap.containsKey(value.getTxId())) {
                    //2.存在的话则输出这个订单对账成功
                    out.collect("订单：" + value.getOrderId() + "对账成功！！");
                    //3.如果存在，提取完了，把txmap中的数据删掉
                    txMap.remove(value.getTxId());
                } else {
                    //不存在，将数据放入orderMap中
                    orderMap.put(value.getTxId(), value);
                }
            }
            @Override
            public void processElement2(TxEvent value, Context ctx, Collector<String> out) throws Exception {
                //1.判断对方缓存中是否有这条数据(TxEvent)能关联上的数据(OrderEvent)
                if (orderMap.containsKey(value.getTxId())) {
                    //2.存在的话则输出这个订单对账成功
                    out.collect("订单：" + orderMap.get(value.getTxId()).getOrderId() + "对账成功！！");
                    //3.如果存在，提取完了，把orderMap中的数据删掉
                    orderMap.remove(value.getTxId());
                } else {
                    //不存在，将数据放入txMap中
                    txMap.put(value.getTxId(), value);
                }
            }
        }).print();

        env.execute();
    }
}
